fix(inkless:controller): skip switched partitions in unfence ISR expansion - #754

…seal

A replica outside ISR must keep fetching from the leader until the leader has
observed the catch-up. The leader records the offset carried by the fetch
request, not the log end offset after the append, so it sees the replica at the
seal only on the following fetch. partitionsAwaitingIsrRecovery exists to
provide that fetch.

The eviction check short-circuited on isConsolidatingPartition and skipped the
wait. Once a consolidating partition hands off, the consolidation fetcher reads
object storage and sends no fetch to the leader, so nothing else expands ISR.
Since #754 also stops the controller from expanding ISR for switched
partitions, a consolidating switched replica outside ISR had no admission path
at all. A broker restart is enough to trigger this, so a rolling restart reaches
every partition of a switched topic: the shutdown drops the replica from ISR,
and after the restart it never rejoins. UnderReplicatedPartitions stays non-zero
and no later restart recovers it, which is the symptom #697 set out to clear.

Drop the short-circuit. It only changed behavior for a replica outside ISR,
which is the case that needs the wait; a replica already in ISR still evicts on
isReplicaInIsr and hands off unchanged.

This matters now because diskless.remote.storage.consolidation.enable requires
diskless.allow.from.classic.enable, so any cluster that enables consolidation
can switch topics, and the switch auto-enables remote storage on the topic.
Every switched partition in such a cluster is consolidating.

Testing

shouldDelayConsolidatingPartitionAtSealWhileOutsideIsr asserts the partition is
queued in partitionsAwaitingIsrRecovery, not evicted, then that the back-off is
applied with replica.fetch.backoff.ms once doWork drains the queue. It is
standalone rather than routed through verifyDisklessSwitchEviction, which proves
only the absence of eviction: that helper never adds the partition to the
fetcher, so delayPartitions is a silent no-op, and doWork clears the queue before
the helper inspects it. Verified red-then-green: restoring the short-circuit
fails the queueing assertion.

…ISR admits it

isReadyForConsolidation routes on offsets alone, so a follower whose local log
already covers the seal goes straight to the consolidation fetcher. That fetcher
reads object storage through DisklessLeaderEndPoint and sends no fetch to the
leader, so the leader never observes the catch-up and the AlterPartition handshake
that expands ISR never starts. Since #754 also stops the controller expanding ISR
for switched partitions, such a replica has no admission path.

A broker restart reaches this state without any leader change: controlled shutdown
removes the replica from ISR, the restart recovers a log end offset at or beyond
the seal, and the routing hands the partition to consolidation before it can prove
itself. UnderReplicatedPartitions then stays non-zero for the rest of the
partition's life, and a rolling restart reaches every partition of a switched
topic. The ISR wait added earlier in this branch cannot help, because the
partition never reaches the classic fetcher where that wait lives.

Hold back a follower sitting exactly at the seal while it is outside ISR. The
classic fetcher waits there until the leader records the position, then evicts and
hands off as it already does for a follower that arrives from below the seal.

Only log end offset == seal qualifies. Past the seal the local latest epoch is the
diskless one, and the classic fetcher's divergence check against the leader's epoch
cache is not built for it, so a partition that has already consolidated stays with
the consolidation fetcher.

Testing

testApplyDeltaKeepsConsolidatingFollowerAtSealOnClassicFetcherWhileOutsideIsr
asserts the classic fetcher receives the partition with its initial offset at the
seal. It captures the argument rather than counting invocations, so an empty map
cannot satisfy it. Without the hold-back the classic fetcher receives nothing.

…seal

A replica outside ISR must keep fetching from the leader until the leader has
observed the catch-up. The leader records the offset carried by the fetch
request, not the log end offset after the append, so it sees the replica at the
seal only on the following fetch. partitionsAwaitingIsrRecovery exists to
provide that fetch.

The eviction check short-circuited on isConsolidatingPartition and skipped the
wait. Once a consolidating partition hands off, the consolidation fetcher reads
object storage and sends no fetch to the leader, so nothing else expands ISR.
Since #754 also stops the controller from expanding ISR for switched
partitions, a consolidating switched replica outside ISR had no admission path
at all. This commit covers a replica that arrives at the seal from below; a
restarted replica never reaches this fetcher at all, which the routing change
later in this branch fixes.

Drop the short-circuit. It only changed behavior for a replica outside ISR,
which is the case that needs the wait; a replica already in ISR still evicts on
isReplicaInIsr and hands off unchanged.

This matters now because diskless.remote.storage.consolidation.enable requires
diskless.allow.from.classic.enable, so any cluster that enables consolidation
can switch topics, and the switch auto-enables remote storage on the topic.
Every switched partition in such a cluster is consolidating.

Testing

shouldDelayConsolidatingPartitionAtSealWhileOutsideIsr asserts the partition is
queued in partitionsAwaitingIsrRecovery, not evicted, then that the back-off is
applied with replica.fetch.backoff.ms once doWork drains the queue. It is
standalone rather than routed through verifyDisklessSwitchEviction, which proves
only the absence of eviction: that helper never adds the partition to the
fetcher, so delayPartitions is a silent no-op, and doWork clears the queue before
the helper inspects it. Verified red-then-green: restoring the short-circuit
fails the queueing assertion.

…ISR admits it

isReadyForConsolidation routes on offsets alone, so a follower whose local log
already covers the seal goes straight to the consolidation fetcher. That fetcher
reads object storage through DisklessLeaderEndPoint and sends no fetch to the
leader, so the leader never observes the catch-up and the AlterPartition handshake
that expands ISR never starts. Since #754 also stops the controller expanding ISR
for switched partitions, such a replica has no admission path.

A broker restart reaches this state without any leader change: controlled shutdown
removes the replica from ISR, the restart recovers a log end offset at or beyond
the seal, and the routing hands the partition to consolidation before it can prove
itself. UnderReplicatedPartitions then stays non-zero for the rest of the
partition's life, and a rolling restart reaches every partition of a switched
topic. The ISR wait added earlier in this branch cannot help, because the
partition never reaches the classic fetcher where that wait lives.

Hold back a follower sitting exactly at the seal while it is outside ISR. The
classic fetcher waits there until the leader admits it back to ISR, then evicts
and hands off as it already does for a follower that arrives from below the seal.
Admission needs the seal-gated predicate added later in this branch: recording the
position alone cannot satisfy the generic in-sync check on a consolidating leader,
whose local high watermark sits at the diskless frontier.

Only log end offset == seal qualifies. Past the seal the local latest epoch is the
diskless one, and the classic fetcher's divergence check against the leader's epoch
cache is not built for it, so a partition that has already consolidated stays with
the consolidation fetcher.

Testing

Three tests pin the routing decision, one per branch of the condition, each failing
for its own mutation:

testApplyDeltaKeepsConsolidatingFollowerAtSealOnClassicFetcherWhileOutsideIsr
asserts the classic fetcher receives the partition with its initial offset at the
seal. It captures the argument rather than counting invocations, so an empty map
cannot satisfy it. Without the hold-back the classic fetcher receives nothing.

testApplyDeltaSendsConsolidatingFollowerAtSealToConsolidationWhileInIsr covers the
same state with the replica in ISR. Dropping the ISR term from the condition parks
an in-ISR replica on the classic fetcher permanently, and fails only this test.

testApplyDeltaSendsConsolidatingFollowerPastSealToConsolidationWhileOutsideIsr
covers a replica already consolidated past the seal. Relaxing log end offset ==
seal to >= sends it back to the classic fetcher and into the divergence check that
cannot read a diskless local epoch, and fails only this test.

…eader high watermark

The at-seal fetch records the follower's position and then relies on the generic
ISR expansion that updateFetchState triggers. That expansion gates on
isFollowerInSync, which compares the recorded offset against this leader's local
high watermark. For a non-consolidating switched partition the leader's log is
frozen at the seal, so the comparison is seal >= seal and the follower joins ISR.

A consolidating leader runs its own consolidation fetcher, and
ReplicaFetcherThread.processPartitionData calls maybeUpdateHighWatermark with the
diskless high watermark, which sets the local watermark directly and bypasses
maybeIncrementLeaderHW. The leader's high watermark is therefore the consolidated
frontier. The at-seal branch deliberately clamps the recorded follower offset to
the seal, because a follower must never replicate diskless records over the
inter-broker path, so the comparison becomes seal >= frontier and is never true.
The follower never joins ISR.

With the hold-back earlier in this branch that is worse than leaving the
short-circuit in place: the follower stays on the classic fetcher, so it is both
under-replicated and never consolidating, and its local log stays pinned at the
seal.

Admit on the seal instead. A fetch at the seal is the whole evidence a switched
partition needs, since the records below it are the only ones that live in local
logs, and it is the same evidence that let #754 drop the controller shortcut.
maybeExpandIsrAtSeal mirrors maybeExpandIsr, replacing the in-sync check with the
seal comparison and keeping everything else: the read-locked fast path, the
in-flight check, the re-check under the write lock and the submit outside it. It is
private[kafka], like the expansion helpers next to it.

The diverging-epoch guard at the call site is deliberate rather than load-bearing.
fetchRecords already leaves the recorded offset at UNKNOWN_OFFSET on divergence, so
the seal comparison fails anyway, but the guard is the clause standing in for the
leaderEpochStartOffsetOpt check that seal-based admission drops, and it should not
rest on a side effect two calls away.

Testing

testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse now models the
leader's watermark at the frontier and asserts an AlterPartition carrying the
follower. Without the frontier watermark the test passed while the production path
could not expand ISR, which is how the branch reached review with the chain
broken. Removing the admission call fails it: no interaction with
alterPartitionManager.

…seal

A replica outside ISR must keep fetching from the leader until the leader has
observed the catch-up. The leader records the offset carried by the fetch
request, not the log end offset after the append, so it sees the replica at the
seal only on the following fetch. partitionsAwaitingIsrRecovery exists to
provide that fetch.

The eviction check short-circuited on isConsolidatingPartition and skipped the
wait. Once a consolidating partition hands off, the consolidation fetcher reads
object storage and sends no fetch to the leader, so nothing else expands ISR.
Since #754 also stops the controller from expanding ISR for switched
partitions, a consolidating switched replica outside ISR had no admission path
at all. This commit covers a replica that arrives at the seal from below; a
restarted replica never reaches this fetcher at all, which the routing change
later in this branch fixes.

Drop the short-circuit. It only changed behavior for a replica outside ISR,
which is the case that needs the wait; a replica already in ISR still evicts on
isReplicaInIsr and hands off unchanged.

This matters now because diskless.remote.storage.consolidation.enable requires
diskless.allow.from.classic.enable, so any cluster that enables consolidation
can switch topics, and the switch auto-enables remote storage on the topic.
Every switched partition in such a cluster is consolidating.

Testing

shouldDelayConsolidatingPartitionAtSealWhileOutsideIsr asserts the partition is
queued in partitionsAwaitingIsrRecovery, not evicted, then that the back-off is
applied with replica.fetch.backoff.ms once doWork drains the queue. It is
standalone rather than routed through verifyDisklessSwitchEviction, which proves
only the absence of eviction: that helper never adds the partition to the
fetcher, so delayPartitions is a silent no-op, and doWork clears the queue before
the helper inspects it. Verified red-then-green: restoring the short-circuit
fails the queueing assertion.

…ISR admits it

isReadyForConsolidation routes on offsets alone, so a follower whose local log
already covers the seal goes straight to the consolidation fetcher. That fetcher
reads object storage through DisklessLeaderEndPoint and sends no fetch to the
leader, so the leader never observes the catch-up and the AlterPartition handshake
that expands ISR never starts. Since #754 also stops the controller expanding ISR
for switched partitions, such a replica has no admission path.

A broker restart reaches this state without any leader change: controlled shutdown
removes the replica from ISR, the restart recovers a log end offset at or beyond
the seal, and the routing hands the partition to consolidation before it can prove
itself. A running broker can reach the same state after fencing: the ISR-only delta
advances the partition epoch but not the leader epoch, so applyLocalFollowersDelta
must explicitly move the existing consolidation fetcher back to the classic fetcher.
Stopping both fetcher managers before the restart keeps that hand-off exclusive.

Without the hold-back and the live reroute, the replica stays out of ISR for the rest
of the partition's life, and a rolling restart reaches every partition of a switched
topic. Nothing reports it: underReplicatedPartitionCount excludes consolidating
partitions, so no metric moves, and the state surfaces only when a leader election
needs an in-sync replica and finds the leader alone. The ISR wait added earlier in
this branch cannot help either, because the partition never reaches the classic
fetcher where that wait lives.

Hold back every follower at or above the seal while it is outside ISR. The classic
fetcher waits there until the leader admits it back to ISR, then evicts and hands off
as it already does for a follower that arrives from below the seal. Admission needs
the seal-gated predicate added later in this branch: recording the position alone
cannot satisfy the generic in-sync check on a consolidating leader, whose local high
watermark sits at the diskless frontier.

A replica that already consolidated past the seal keeps its diskless suffix while it
waits. initialFetchOffset starts the classic fetcher at the real log end offset, and
the leader answers from the at-seal branch rather than its local log, so nothing
overwrites the suffix. The epoch the follower reports is the diskless one, and it
resolves in the leader's epoch cache whenever the leader has consolidated at least as
far, which is the ordinary case: the fetch reports no divergence and the follower is
admitted with its suffix intact.

When the leader has consolidated nothing, its cache cannot resolve that epoch and the
fetch answers OFFSET_OUT_OF_RANGE. The follower recovers through the standard path: a
follower ListOffsets reads the leader's local log, so the latest offset it gets back is
the seal, and it truncates there and is admitted on the next fetch. Losing the suffix
is unavoidable in that case, because this leader does not hold it locally.

Restricting the hold-back to log end offset == seal would have been narrower and wrong.
A follower that consolidated anything at all sits above the seal, so the common
rolling-restart case would still route to consolidation and still have no way back into
ISR.

Testing

Three tests pin the routing decision, one per branch of the condition, each failing
for its own mutation:

testApplyDeltaKeepsConsolidatingFollowerAtSealOnClassicFetcherWhileOutsideIsr
asserts the classic fetcher receives the partition with its initial offset at the
seal. It captures the argument rather than counting invocations, so an empty map
cannot satisfy it. Without the hold-back the classic fetcher receives nothing.

testApplyDeltaKeepsConsolidatingFollowerPastSealOnClassicFetcherWhileOutsideIsr
covers a replica already consolidated past the seal: it asserts the classic fetcher
takes it at the real log end offset, not at the seal, and that the local log still
ends where it did. Restoring the seal equality fails only this test.

testApplyDeltaSendsConsolidatingFollowerAtSealToConsolidationWhileInIsr covers the
same state with the replica in ISR. Dropping the ISR term from the condition parks
an in-ISR replica on the classic fetcher permanently, and fails only that test.

testApplyDeltaReroutesConsolidatingFollowerToClassicFetcherWhenRemovedFromIsr
applies an ISR-only delta with the same leader epoch and a higher partition epoch. It
asserts that the partition starts on consolidation, both fetcher managers remove it,
and the classic fetcher restarts at the existing log end offset.

Two tests cover the epoch reconciliation the hold-back now depends on.
testFollowerFetchPastSealCarryingDisklessEpochIsAdmittedWhenLeaderHasConsolidated
asserts a fetch above the seal carrying the diskless epoch reports no divergence and
produces an AlterPartition, with both logs intact.
testFollowerFetchPastSealSignalsOutOfRangeWhenLeaderLacksTheDisklessEpoch pins the
other leader state, where the answer is OFFSET_OUT_OF_RANGE and no AlterPartition.

…eader high watermark

The at-seal fetch records the follower's position and then relies on the generic
ISR expansion that updateFetchState triggers. That expansion gates on
isFollowerInSync, which compares the recorded offset against this leader's local
high watermark. For a non-consolidating switched partition the leader's log is
frozen at the seal, so the comparison is seal >= seal and the follower joins ISR.

A consolidating leader runs its own consolidation fetcher, and
ReplicaFetcherThread.processPartitionData calls maybeUpdateHighWatermark with the
diskless high watermark, which sets the local watermark directly and bypasses
maybeIncrementLeaderHW. The leader's high watermark is therefore the consolidated
frontier. The at-seal branch deliberately clamps the recorded follower offset to
the seal, because a follower must never replicate diskless records over the
inter-broker path, so the comparison becomes seal >= frontier and is never true.
The follower never joins ISR.

With the hold-back earlier in this branch that is worse than leaving the
short-circuit in place: the follower stays on the classic fetcher, so it is both
under-replicated and never consolidating, and its local log stays pinned at the
seal.

Admit on the seal instead. A fetch at the seal is the whole evidence a switched
partition needs, since the records below it are the only ones that live in local
logs, and it is the same evidence that let #754 drop the controller shortcut.
maybeExpandIsrAtSeal mirrors maybeExpandIsr, replacing the in-sync check with the
seal comparison and keeping everything else: the read-locked fast path, the
in-flight check, the re-check under the write lock and the submit outside it. It is
private[kafka], like the expansion helpers next to it.

The diverging-epoch guard at the call site is deliberate rather than load-bearing.
fetchRecords already leaves the recorded offset at UNKNOWN_OFFSET on divergence, so
the seal comparison fails anyway, but the guard is the clause standing in for the
leaderEpochStartOffsetOpt check that seal-based admission drops, and it should not
rest on a side effect two calls away.

Testing

testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse now models the
leader's watermark at the frontier and asserts an AlterPartition carrying the
follower. Without the frontier watermark the test passed while the production path
could not expand ISR, which is how the branch reached review with the chain
broken. Removing the admission call fails it: no interaction with
alterPartitionManager.

Two further tests cover the epoch reconciliation that admission depends on once a
follower above the seal reaches this branch.
testFollowerFetchPastSealCarryingDisklessEpochIsAdmittedWhenLeaderHasConsolidated
asserts a fetch above the seal carrying the diskless epoch reports no divergence and
produces an AlterPartition, with both logs left intact.
testFollowerFetchPastSealSignalsOutOfRangeWhenLeaderLacksTheDisklessEpoch pins the
other leader state: the answer is OFFSET_OUT_OF_RANGE and no AlterPartition, which is
what sends the follower down out-of-range recovery.

…seal

A replica outside ISR must keep fetching from the leader until the leader has
observed the catch-up. The leader records the offset carried by the fetch
request, not the log end offset after the append, so it sees the replica at the
seal only on the following fetch. partitionsAwaitingIsrRecovery exists to
provide that fetch.

The eviction check short-circuited on isConsolidatingPartition and skipped the
wait. Once a consolidating partition hands off, the consolidation fetcher reads
object storage and sends no fetch to the leader, so nothing else expands ISR.
Since #754 also stops the controller from expanding ISR for switched
partitions, a consolidating switched replica outside ISR had no admission path
at all. This commit covers a replica that arrives at the seal from below; a
restarted replica never reaches this fetcher at all, which the routing change
later in this branch fixes.

Drop the short-circuit. It only changed behavior for a replica outside ISR,
which is the case that needs the wait; a replica already in ISR still evicts on
isReplicaInIsr and hands off unchanged.

This matters now because diskless.remote.storage.consolidation.enable requires
diskless.allow.from.classic.enable, so any cluster that enables consolidation
can switch topics, and the switch auto-enables remote storage on the topic.
Every switched partition in such a cluster is consolidating.

Testing

shouldDelayConsolidatingPartitionAtSealWhileOutsideIsr asserts the partition is
queued in partitionsAwaitingIsrRecovery, not evicted, then that the back-off is
applied with replica.fetch.backoff.ms once doWork drains the queue. It is
standalone rather than routed through verifyDisklessSwitchEviction, which proves
only the absence of eviction: that helper never adds the partition to the
fetcher, so delayPartitions is a silent no-op, and doWork clears the queue before
the helper inspects it. Verified red-then-green: restoring the short-circuit
fails the queueing assertion.

…ISR admits it

isReadyForConsolidation routes on offsets alone, so a follower whose local log
already covers the seal goes straight to the consolidation fetcher. That fetcher
reads object storage through DisklessLeaderEndPoint and sends no fetch to the
leader, so the leader never observes the catch-up and the AlterPartition handshake
that expands ISR never starts. Since #754 also stops the controller expanding ISR
for switched partitions, such a replica has no admission path.

A broker restart reaches this state without any leader change: controlled shutdown
removes the replica from ISR, the restart recovers a log end offset at or beyond
the seal, and the routing hands the partition to consolidation before it can prove
itself. A running broker can reach the same state after fencing: the ISR-only delta
advances the partition epoch but not the leader epoch, so applyLocalFollowersDelta
must explicitly move the existing consolidation fetcher back to the classic fetcher.
Stopping both fetcher managers before the restart keeps that hand-off exclusive.

Without the hold-back and the live reroute, the replica stays out of ISR for the rest
of the partition's life, and a rolling restart reaches every partition of a switched
topic. Nothing reports it: underReplicatedPartitionCount excludes consolidating
partitions, so no metric moves, and the state surfaces only when a leader election
needs an in-sync replica and finds the leader alone. The ISR wait added earlier in
this branch cannot help either, because the partition never reaches the classic
fetcher where that wait lives.

Hold back every follower at or above the seal while it is outside ISR. The classic
fetcher waits there until the leader admits it back to ISR, then evicts and hands off
as it already does for a follower that arrives from below the seal. Admission needs
the seal-gated predicate added later in this branch: recording the position alone
cannot satisfy the generic in-sync check on a consolidating leader, whose local high
watermark sits at the diskless frontier.

A replica that already consolidated past the seal keeps its diskless suffix while it
waits. initialFetchOffset starts the classic fetcher at the real log end offset, and
the leader answers from the at-seal branch rather than its local log, so nothing
immediately overwrites the suffix. The later admission commit validates that state
against a leader that holds the complete classic prefix. If that leader cannot
resolve the diskless epoch, standard out-of-range recovery may discard only the
post-seal suffix. A leader below the seal is different: it cannot safely validate or
truncate the follower until it rebuilds its own classic prefix, which the admission
commit also guards.

Restricting the hold-back to log end offset == seal would have been narrower and wrong.
A follower that consolidated anything at all sits above the seal, so the common
rolling-restart case would still route to consolidation and still have no way back into
ISR.

Testing

Four tests pin the routing decision and the live reroute:

testApplyDeltaKeepsConsolidatingFollowerAtSealOnClassicFetcherWhileOutsideIsr
asserts the classic fetcher receives the partition with its initial offset at the
seal. It captures the argument rather than counting invocations, so an empty map
cannot satisfy it. Without the hold-back the classic fetcher receives nothing.

testApplyDeltaKeepsConsolidatingFollowerPastSealOnClassicFetcherWhileOutsideIsr
covers a replica already consolidated past the seal: it asserts the classic fetcher
takes it at the real log end offset, not at the seal, and that the local log still
ends where it did. Restoring the seal equality fails only this test.

testApplyDeltaSendsConsolidatingFollowerAtSealToConsolidationWhileInIsr covers the
same state with the replica in ISR. Dropping the ISR term from the condition parks
an in-ISR replica on the classic fetcher permanently, and fails only this test.

testApplyDeltaReroutesConsolidatingFollowerToClassicFetcherWhenRemovedFromIsr
applies an ISR-only delta with the same leader epoch and a higher partition epoch. It
asserts that the partition starts on consolidation, both fetcher managers remove it,
and the classic fetcher restarts at the existing log end offset.

…nsion (#754)

expandIsrForDisklessManagedPartitions re-admitted a returning broker to the ISR
of every diskless.enable=true partition, selecting on the topic config alone.
That is correct for born-diskless partitions, the case #643 targeted: all their
data is in object storage, so any live replica is current.

A partition switched from classic breaks that premise. Its records below
classicToDisklessStartOffset exist only in the replicas' local logs, so a
returning replica could hold an incomplete prefix and was still placed in ISR
and made electable.

Skip any partition whose seal is not NO_CLASSIC_TO_DISKLESS_START_OFFSET, which
covers both a committed seal and a switch still PENDING. Those partitions earn
ISR through AlterPartition once the leader observes follower fetch state at the
seal -- the path #697 added. Removing this shortcut before that path existed
would have left switched partitions with no ISR-recovery route at all.

The test drives one unfence across three topics -- born-diskless, switched with
a committed seal, and switch-pending. Asserting only that switched partitions
stay out would also pass against a guard that skipped everything, so the
born-diskless leg is what makes it discriminating. Verified red-then-green: with
the guard reverted, the switched assertion fails with expected <false> but was
<true>.

…nsion (#754)

expandIsrForDisklessManagedPartitions re-admitted a returning broker to the ISR
of every diskless.enable=true partition, selecting on the topic config alone.
That is correct for born-diskless partitions, the case #643 targeted: all their
data is in object storage, so any live replica is current.

A partition switched from classic breaks that premise. Its records below
classicToDisklessStartOffset exist only in the replicas' local logs, so a
returning replica could hold an incomplete prefix and was still placed in ISR
and made electable.

Skip any partition whose seal is not NO_CLASSIC_TO_DISKLESS_START_OFFSET, which
covers both a committed seal and a switch still PENDING. Those partitions earn
ISR through AlterPartition once the leader observes follower fetch state at the
seal -- the path #697 added. Removing this shortcut before that path existed
would have left switched partitions with no ISR-recovery route at all.

The test drives one unfence across three topics -- born-diskless, switched with
a committed seal, and switch-pending. Asserting only that switched partitions
stay out would also pass against a guard that skipped everything, so the
born-diskless leg is what makes it discriminating. Verified red-then-green: with
the guard reverted, the switched assertion fails with expected <false> but was
<true>.

…seal

A replica outside ISR must keep fetching from the leader until the leader has
observed the catch-up. The leader records the offset carried by the fetch
request, not the log end offset after the append, so it sees the replica at the
seal only on the following fetch. partitionsAwaitingIsrRecovery exists to
provide that fetch.

The eviction check short-circuited on isConsolidatingPartition and skipped the
wait. Once a consolidating partition hands off, the consolidation fetcher reads
object storage and sends no fetch to the leader, so nothing else expands ISR.
Since #754 also stops the controller from expanding ISR for switched
partitions, a consolidating switched replica outside ISR had no admission path
at all. This commit covers a replica that arrives at the seal from below; a
restarted replica never reaches this fetcher at all, which the routing change
later in this branch fixes.

Drop the short-circuit. It only changed behavior for a replica outside ISR,
which is the case that needs the wait; a replica already in ISR still evicts on
isReplicaInIsr and hands off unchanged.

The pending backoff belongs to one fetcher residency. Partition removal can run on
the metadata publisher thread, so remove the pending entry under partitionMapLock
and drain the buffer under the same lock. Fetcher managers hold their monitor before
calling removePartitions, so never call replicaFetcherManager or
consolidationFetcherManager while holding partitionMapLock; that would invert the
manager-to-partition order.

This matters now because diskless.remote.storage.consolidation.enable requires
diskless.allow.from.classic.enable, so any cluster that enables consolidation
can switch topics, and the switch auto-enables remote storage on the topic.
Every switched partition in such a cluster is consolidating.

Testing

shouldDelayConsolidatingPartitionAtSealWhileOutsideIsr asserts the partition is
queued in partitionsAwaitingIsrRecovery, not evicted, then that the back-off is
applied with replica.fetch.backoff.ms once doWork drains the queue. It is
standalone rather than routed through verifyDisklessSwitchEviction, which proves
only the absence of eviction: that helper never adds the partition to the
fetcher, so delayPartitions is a silent no-op, and doWork clears the queue before
the helper inspects it. Verified red-then-green: restoring the short-circuit
fails the queueing assertion.

shouldClearPendingIsrRecoveryBackoffWhenPartitionIsRemoved removes and re-adds a
partition before the pending buffer drains. The new residency remains immediately
fetchable instead of inheriting the old residency's delay.

…ISR admits it

isReadyForConsolidation routes on offsets alone, so a follower whose local log
already covers the seal goes straight to the consolidation fetcher. That fetcher
reads object storage through DisklessLeaderEndPoint and sends no fetch to the
leader, so the leader never observes the catch-up and the AlterPartition handshake
that expands ISR never starts. Since #754 also stops the controller expanding ISR
for switched partitions, such a replica has no admission path.

A broker restart reaches this state without any leader change: controlled shutdown
removes the replica from ISR, the restart recovers a log end offset at or beyond
the seal, and the routing hands the partition to consolidation before it can prove
itself. A running broker can reach the same state after fencing: the ISR-only delta
advances the partition epoch but not the leader epoch, so applyLocalFollowersDelta
must explicitly move the existing consolidation fetcher back to the classic fetcher.
Stopping both fetcher managers before the restart keeps that hand-off exclusive.

Without the hold-back and the live reroute, the replica stays out of ISR for the rest
of the partition's life, and a rolling restart reaches every partition of a switched
topic. Nothing reports it: underReplicatedPartitionCount excludes consolidating
partitions, so no metric moves, and the state surfaces only when a leader election
needs an in-sync replica and finds the leader alone. The ISR wait added earlier in
this branch cannot help either, because the partition never reaches the classic
fetcher where that wait lives.

Hold back every follower at or above the seal while it is outside ISR. The classic
fetcher waits there until the leader admits it back to ISR, then evicts and hands off
as it already does for a follower that arrives from below the seal. Admission needs
the seal-gated predicate added later in this branch: recording the position alone
cannot satisfy the generic in-sync check on a consolidating leader, whose local high
watermark sits at the diskless frontier.

A replica that already consolidated past the seal keeps its diskless suffix while it
waits. initialFetchOffset starts the classic fetcher at the real log end offset, and
the leader answers from the at-seal branch rather than its local log, so nothing
immediately overwrites the suffix. The later admission commit validates that state
against a leader that holds the complete classic prefix. If that leader cannot
resolve the diskless epoch, standard out-of-range recovery may discard only the
post-seal suffix. A leader below the seal is different: it cannot safely validate or
truncate the follower until it rebuilds its own classic prefix, which the admission
commit also guards.

Restricting the hold-back to log end offset == seal would have been narrower and wrong.
A follower that consolidated anything at all sits above the seal, so the common
rolling-restart case would still route to consolidation and still have no way back into
ISR.

Testing

Four tests pin the routing decision and the live reroute:

testApplyDeltaKeepsConsolidatingFollowerAtSealOnClassicFetcherWhileOutsideIsr
asserts the classic fetcher receives the partition with its initial offset at the
seal. It captures the argument rather than counting invocations, so an empty map
cannot satisfy it. Without the hold-back the classic fetcher receives nothing.

testApplyDeltaKeepsConsolidatingFollowerPastSealOnClassicFetcherWhileOutsideIsr
covers a replica already consolidated past the seal: it asserts the classic fetcher
takes it at the real log end offset, not at the seal, and that the local log still
ends where it did. Restoring the seal equality fails only this test.

testApplyDeltaSendsConsolidatingFollowerAtSealToConsolidationWhileInIsr covers the
same state with the replica in ISR. Dropping the ISR term from the condition parks
an in-ISR replica on the classic fetcher permanently, and fails only this test.

testApplyDeltaReroutesConsolidatingFollowerToClassicFetcherWhenRemovedFromIsr
applies an ISR-only delta with the same leader epoch and a higher partition epoch. It
asserts that the partition starts on consolidation, both fetcher managers remove it,
and the classic fetcher restarts at the existing log end offset.